package com.xiva.zk.curator;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;

import java.net.URL;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Description Curator Demo
 * @Author XIVA
 * @Date 2019/9/10
 */
public class CuratorZookeeperClient
{
    private final CuratorFramework client;

    private int sessionTimeout = 5000;

    static final Charset CHARSET = Charset.forName("UTF-8");

    private String connectString = "192.168.88.129:2181";

    /**
     * 连接zk服务
     */
    public CuratorZookeeperClient()
    {
        try
        {
            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(connectString).retryPolicy(new RetryNTimes(1, 1000)).connectionTimeoutMs(5000);

            client = builder.build();

            client.start();
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 创建永久节点
     * @param path
     */
    public void createPersistent(String path)
    {
        try
        {
            client.create().forPath(path);
        }
        catch (KeeperException.NodeExistsException e)
        {
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 创建临时节点
     * @param path
     */
    public void createEphemeral(String path)
    {
        try
        {
            client.create().withMode(CreateMode.EPHEMERAL).forPath(path);
        }
        catch (KeeperException.NodeExistsException e)
        {
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 创建永久节点，并写入数据
     * @param path
     * @param data
     */
    public void createPersistent(String path, String data)
    {
        byte[] dataBytes = data.getBytes(CHARSET);
        this.createPersistent(path, dataBytes);
    }

    /**
     * 创建永久节点，写入byte[]s数据
     * @param path
     * @param dataBytes
     */
    public void createPersistent(String path, byte[] dataBytes)
    {
        try
        {
            client.create().forPath(path, dataBytes);
        }
        catch (KeeperException.NodeExistsException e)
        {
            try
            {
                client.setData().forPath(path, dataBytes);
            }
            catch (Exception e1)
            {
                throw new IllegalStateException(e.getMessage(), e1);
            }
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 创建临时节点，并写入数据
     * @param path
     * @param data
     */
    public void createEphemeral(String path, String data)
    {
        byte[] dataBytes = data.getBytes(CHARSET);
        try
        {
            client.create().withMode(CreateMode.EPHEMERAL).forPath(path, dataBytes);
        }
        catch (KeeperException.NodeExistsException e)
        {
            try
            {
                client.setData().forPath(path, dataBytes);
            }
            catch (Exception e1)
            {
                throw new IllegalStateException(e.getMessage(), e1);
            }
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 创建临时 序列编号节点，会递增；用来选举用，可选举最小值为master。
     * 资源命名作为命名目录服务
     * @param path
     * @param data
     */
    public void createEphemeralSequential(String path, String data)
    {
        byte[] dataBytes = data.getBytes(CHARSET);
        try
        {
            client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(path, dataBytes);
        }
        catch (KeeperException.NodeExistsException e)
        {
            try
            {
                client.setData().forPath(path, dataBytes);
            }
            catch (Exception e1)
            {
                throw new IllegalStateException(e.getMessage(), e1);
            }
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 删除节点
     * @param path
     */
    public void delete(String path)
    {
        try
        {
            client.delete().forPath(path);
        }
        catch (KeeperException.NoNodeException e)
        {
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 获取子节点
     * @param path
     * @return
     */
    public List<String> getChildren(String path)
    {
        try
        {
            return client.getChildren().forPath(path);
        }
        catch (KeeperException.NoNodeException e)
        {
            return null;
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
    }

    /**
     * 判断节点是否存在
     * @param path
     * @return
     */
    public boolean checkExists(String path)
    {
        try
        {
            if (client.checkExists().forPath(path) != null)
            {
                return true;
            }
        }
        catch (Exception e)
        {
        }
        return false;
    }

    /**
     * 判断是否连接
     * @return
     */
    public boolean isConnected()
    {
        return client.getZookeeperClient().isConnected();
    }

    /**
     * 获取路径下内容
     * @param path
     * @return
     */
    public String doGetContent(String path)
    {
        try
        {
            byte[] dataBytes = client.getData().forPath(path);
            return (dataBytes == null || dataBytes.length == 0) ? null : new String(dataBytes, CHARSET);
        }
        catch (KeeperException.NoNodeException e)
        {
            // ignore NoNode Exception.
        }
        catch (Exception e)
        {
            throw new IllegalStateException(e.getMessage(), e);
        }
        return null;
    }

    /**
     * 关闭客户端
     */
    public void doClose()
    {
        client.close();
    }

    public int getSessionTimeout()
    {
        return sessionTimeout;
    }
}
